package com.campus.delivery.service;

import com.campus.common.message.OrderCompleteMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.math.BigDecimal;

/**
 * 订单完成消息生产者（配送端）
 * 负责向Kafka发送订单完成消息
 */
@Service
@Slf4j
public class OrderCompleteMessageProducer {

    private static final String TOPIC_ORDER_COMPLETE = "order-complete";

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Resource
    private ObjectMapper objectMapper;

    /**
     * 发送订单完成消息
     * @param openId 用户openId
     * @param orderId 订单号
     * @param goodsName 货品名称
     * @param orderAmount 订单金额
     * @param completeTime 完成时间
     * @param remark 订单备注
     */
    public void sendOrderCompleteMessage(String openId, String orderId, String goodsName, BigDecimal orderAmount, String completeTime, String remark) {
        try {
            OrderCompleteMessage message = new OrderCompleteMessage(openId, orderId, goodsName, orderAmount, completeTime, remark);

            String messageJson = objectMapper.writeValueAsString(message);
            
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_ORDER_COMPLETE, orderId, messageJson);
            
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    log.info("订单完成消息发送成功，订单号: {}, 用户openId: {}, offset: {}", 
                            orderId, openId, result.getRecordMetadata().offset());
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.error("订单完成消息发送失败，订单号: {}, 用户openId: {}", orderId, openId, ex);
                }
            });
        } catch (Exception e) {
            log.error("发送订单完成消息异常，订单号: {}, 用户openId: {}", orderId, openId, e);
        }
    }
}